//! Pipeline响应管理器
//! 
//! 确保Redis Pipeline命令的响应按正确顺序返回给客户端

use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};
use std::time::Instant;

use crate::storage::CommandResult;
use crate::event::{ConnectionId, RequestId};
use crate::protocol::{RespParser, ConnectionManager};

/// Pipeline响应项
#[derive(Debug, Clone)]
pub struct PipelineResponse {
    /// 请求ID
    pub request_id: RequestId,
    /// 连接ID
    pub connection_id: ConnectionId,
    /// 命令执行结果
    pub result: CommandResult,
    /// 序列号（用于排序）
    pub sequence_number: u64,
    /// 响应创建时间
    pub created_at: Instant,
}

/// Pipeline响应管理器
/// 
/// 负责：
/// 1. 维护每个连接的响应顺序队列
/// 2. 确保响应按命令接收顺序发送
/// 3. 处理乱序到达的响应
pub struct PipelineManager {
    /// 每个连接的响应队列: ConnectionId -> VecDeque<PipelineResponse>
    connection_queues: Arc<Mutex<HashMap<ConnectionId, VecDeque<PipelineResponse>>>>,
    /// 每个连接期待的下一个序列号: ConnectionId -> u64
    expected_sequences: Arc<Mutex<HashMap<ConnectionId, u64>>>,
    /// 乱序响应缓存: ConnectionId -> HashMap<sequence_number, PipelineResponse>
    pending_responses: Arc<Mutex<HashMap<ConnectionId, HashMap<u64, PipelineResponse>>>>,
    /// 连接管理器引用
    connection_manager: Arc<ConnectionManager>,
    /// RESP解析器（用于序列化响应）
    resp_parser: RespParser,
}

impl PipelineManager {
    /// 创建新的Pipeline管理器
    pub fn new(connection_manager: Arc<ConnectionManager>) -> Self {
        Self {
            connection_queues: Arc::new(Mutex::new(HashMap::new())),
            expected_sequences: Arc::new(Mutex::new(HashMap::new())),
            pending_responses: Arc::new(Mutex::new(HashMap::new())),
            connection_manager,
            resp_parser: RespParser::new(),
        }
    }

    /// 注册新连接的Pipeline管理
    pub fn register_connection(&self, connection_id: ConnectionId) {
        let mut queues = self.connection_queues.lock().unwrap();
        let mut sequences = self.expected_sequences.lock().unwrap();
        let mut pending = self.pending_responses.lock().unwrap();
        
        queues.insert(connection_id, VecDeque::new());
        sequences.insert(connection_id, 0);
        pending.insert(connection_id, HashMap::new());
        
        println!("📋 为连接 {connection_id} 注册Pipeline管理");
    }

    /// 注销连接的Pipeline管理
    pub fn unregister_connection(&self, connection_id: ConnectionId) {
        let mut queues = self.connection_queues.lock().unwrap();
        let mut sequences = self.expected_sequences.lock().unwrap();
        let mut pending = self.pending_responses.lock().unwrap();
        
        queues.remove(&connection_id);
        sequences.remove(&connection_id);
        pending.remove(&connection_id);
        
        println!("📋 注销连接 {connection_id} 的Pipeline管理");
    }

    /// 添加响应（可能乱序到达）
    pub fn add_response(&self, response: PipelineResponse) -> Result<(), String> {
        let connection_id = response.connection_id;
        
        let sequences = self.expected_sequences.lock().unwrap();
        let mut pending = self.pending_responses.lock().unwrap();
        
        let expected_seq = sequences.get(&connection_id).copied().unwrap_or(0);
        
        if response.sequence_number == expected_seq {
            // 这是期待的下一个响应，可以立即发送
            drop(sequences);
            drop(pending);
            
            self.send_response_immediately(&response)?;
            self.advance_sequence(connection_id);
            
            // 检查是否有后续响应可以发送
            self.try_send_pending_responses(connection_id)?;
        } else if response.sequence_number > expected_seq {
            // 响应来得太早，缓存起来
            let pending_map = pending.entry(connection_id).or_default();
            let seq_num = response.sequence_number; // 保存序列号用于日志
            pending_map.insert(response.sequence_number, response);
            
            println!("🔄 缓存乱序响应: 连接:{connection_id} 序列:{seq_num} 期待:{expected_seq}");
        } else {
            // 响应来得太晚，可能是重复或错误
            return Err(format!(
                "收到过期响应: 连接:{} 序列:{} 期待:{}",
                connection_id, response.sequence_number, expected_seq
            ));
        }
        
        Ok(())
    }

    /// 立即发送响应
    fn send_response_immediately(&self, response: &PipelineResponse) -> Result<(), String> {
        let resp_bytes = self.resp_parser.command_result_to_resp(&response.result);
        
        self.connection_manager.write_response(response.connection_id, &resp_bytes)
            .map(|_| ())
            .map_err(|e| format!("发送响应失败: {e}"))?;
        
        println!("📤 发送Pipeline响应: 连接:{} 序列:{} 大小:{}字节", 
            response.connection_id, response.sequence_number, resp_bytes.len());
        
        Ok(())
    }

    /// 前进序列号
    fn advance_sequence(&self, connection_id: ConnectionId) {
        let mut sequences = self.expected_sequences.lock().unwrap();
        if let Some(seq) = sequences.get_mut(&connection_id) {
            *seq += 1;
        }
    }

    /// 尝试发送待处理的响应
    fn try_send_pending_responses(&self, connection_id: ConnectionId) -> Result<(), String> {
        loop {
            let sequences = self.expected_sequences.lock().unwrap();
            let mut pending = self.pending_responses.lock().unwrap();
            
            let expected_seq = sequences.get(&connection_id).copied().unwrap_or(0);
            let pending_map = pending.entry(connection_id).or_default();
            
            if let Some(response) = pending_map.remove(&expected_seq) {
                drop(sequences);
                drop(pending);
                
                // 发送这个响应
                self.send_response_immediately(&response)?;
                self.advance_sequence(connection_id);
                
                // 继续检查下一个
                continue;
            } else {
                // 没有后续响应可发送
                break;
            }
        }
        
        Ok(())
    }

    /// 获取连接的统计信息
    pub fn get_connection_stats(&self, connection_id: ConnectionId) -> PipelineConnectionStats {
        let queues = self.connection_queues.lock().unwrap();
        let sequences = self.expected_sequences.lock().unwrap();
        let pending = self.pending_responses.lock().unwrap();
        
        let queue_size = queues.get(&connection_id).map(|q| q.len()).unwrap_or(0);
        let expected_seq = sequences.get(&connection_id).copied().unwrap_or(0);
        let pending_count = pending.get(&connection_id).map(|p| p.len()).unwrap_or(0);
        
        PipelineConnectionStats {
            connection_id,
            queue_size,
            expected_sequence: expected_seq,
            pending_responses: pending_count,
        }
    }

    /// 获取所有连接的统计信息
    pub fn get_overall_stats(&self) -> PipelineOverallStats {
        let queues = self.connection_queues.lock().unwrap();
        let pending = self.pending_responses.lock().unwrap();
        
        let total_connections = queues.len();
        let total_queued = queues.values().map(|q| q.len()).sum();
        let total_pending = pending.values().map(|p| p.len()).sum();
        
        PipelineOverallStats {
            total_connections,
            total_queued_responses: total_queued,
            total_pending_responses: total_pending,
        }
    }

    /// 清理超时的待处理响应
    pub fn cleanup_timeout_responses(&self, timeout_seconds: u64) -> usize {
        let mut pending = self.pending_responses.lock().unwrap();
        let timeout_duration = std::time::Duration::from_secs(timeout_seconds);
        let now = Instant::now();
        
        let mut cleaned_count = 0;
        
        for (connection_id, pending_map) in pending.iter_mut() {
            let mut to_remove = Vec::new();
            
            for (seq, response) in pending_map.iter() {
                if now.duration_since(response.created_at) > timeout_duration {
                    to_remove.push(*seq);
                }
            }
            
            for seq in to_remove {
                pending_map.remove(&seq);
                cleaned_count += 1;
                println!("🧹 清理超时响应: 连接:{connection_id} 序列:{seq}");
            }
        }
        
        cleaned_count
    }
}

/// Pipeline连接统计
#[derive(Debug, Clone)]
pub struct PipelineConnectionStats {
    pub connection_id: ConnectionId,
    pub queue_size: usize,
    pub expected_sequence: u64,
    pub pending_responses: usize,
}

/// Pipeline整体统计
#[derive(Debug, Clone)]
pub struct PipelineOverallStats {
    pub total_connections: usize,
    pub total_queued_responses: usize,
    pub total_pending_responses: usize,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::protocol::NetworkStream;
    use std::io;

    // 模拟网络流
    #[derive(Debug)]
    struct MockStream;
    impl NetworkStream for MockStream {
        fn read(&mut self, _buf: &mut [u8]) -> io::Result<usize> { Ok(0) }
        fn write(&mut self, buf: &[u8]) -> io::Result<usize> { Ok(buf.len()) }
        fn flush(&mut self) -> io::Result<()> { Ok(()) }
        fn is_connected(&self) -> bool { true }
        fn remote_addr(&self) -> Option<String> { Some("127.0.0.1:12345".to_string()) }
        fn shutdown(&mut self) -> io::Result<()> { Ok(()) }
    }

    #[test]
    fn test_pipeline_ordering() {
        let connection_manager = Arc::new(ConnectionManager::new());
        let pipeline_manager = PipelineManager::new(Arc::clone(&connection_manager));
        
        // 注册连接
        let stream = Box::new(MockStream);
        let conn_id = connection_manager.register_connection(stream);
        pipeline_manager.register_connection(conn_id);
        
        // 创建乱序响应
        let response1 = PipelineResponse {
            request_id: 1,
            connection_id: conn_id,
            result: CommandResult::String("first".to_string()),
            sequence_number: 0,
            created_at: Instant::now(),
        };
        
        let response3 = PipelineResponse {
            request_id: 3,
            connection_id: conn_id,
            result: CommandResult::String("third".to_string()),
            sequence_number: 2,
            created_at: Instant::now(),
        };
        
        let response2 = PipelineResponse {
            request_id: 2,
            connection_id: conn_id,
            result: CommandResult::String("second".to_string()),
            sequence_number: 1,
            created_at: Instant::now(),
        };
        
        // 按乱序添加响应
        pipeline_manager.add_response(response1).unwrap(); // 应该立即发送
        pipeline_manager.add_response(response3).unwrap(); // 应该缓存
        pipeline_manager.add_response(response2).unwrap(); // 应该触发发送response2和response3
        
        // 验证统计
        let stats = pipeline_manager.get_connection_stats(conn_id);
        assert_eq!(stats.expected_sequence, 3); // 期待序列号3
        assert_eq!(stats.pending_responses, 0); // 没有待处理响应
    }
}